package cn.doitedu.hbase.day02;

import cn.doitedu.hbase.day01.ConnectionUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Iterator;

/**
 * 多hbase进行扫描，返回0到多条数据
 */
public class ScanDemo {

    public static void main(String[] args) throws Exception {


        Connection connection = ConnectionUtils.getConnection();
        Table table = connection.getTable(TableName.valueOf("tb_users"));

        //查询指定表中的全部数据
        //scanAll(table);

        //按照rowkey范围查询数据
        //scanByRowKeyRange(table);

        //设置扫描的最大条数
        //scanWithLimit(table);

        //设置扫描rowkey的前缀条件
        //scanWithRowKeyPrefix(table);

        //扫描指定的一个或多个列族的数据
        //scanWithFamily(table);

        //扫描指定的一个列和列标识符的值进行过滤
        //scanWithColumnValue(table);

        //按照多个条件进行过滤
        scanWithFilters(table);


        table.close();
        connection.close();

    }

    private static void scanAll(Table table) throws Exception {

        //查询多条数据
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();
        printData(iterator);

    }


    private static void scanByRowKeyRange(Table table) throws Exception {

        //查询多条数据
        Scan scan = new Scan();
        //添加条件（按照rowkey的范围扫描）
        //默认情况下，按照rowkey范围扫描，是一个前闭后开的区间["u010", "u013")
        //scan.withStartRow(Bytes.toBytes("u010")); //设置行键的起始行
        //scan.withStopRow(Bytes.toBytes("u013")); //设置行键的结束行

        //设置rowkey的范围：("u010", "u013"]
        scan.withStartRow(Bytes.toBytes("u010"), false); //设置行键的起始行
        scan.withStopRow(Bytes.toBytes("u013"), true); //设置行键的结束行


        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();
        printData(iterator);

    }


    private static void scanWithLimit(Table table) throws Exception {

        //查询多条数据
        Scan scan = new Scan();
        //设置rowkey的范围：("u010", "u013"]
        scan.withStartRow(Bytes.toBytes("u010"), true); //设置行键的起始行
        scan.withStopRow(Bytes.toBytes("u013"), true); //设置行键的结束行

        //设置扫描的最大条数
        scan.setLimit(3);

        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();
        printData(iterator);

    }

    /**
     * 按照rowkey的前缀匹配进行扫描
     *
     * @param table
     * @throws Exception
     */
    private static void scanWithRowKeyPrefix(Table table) throws Exception {

        Scan scan = new Scan();
        //指定按照rowkey的前缀进行扫描

        //设置rowkey过滤前缀
        PrefixFilter filter = new PrefixFilter(Bytes.toBytes("u01"));
        //将扫描的过滤器关联带scan上
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        printData(iterator);


    }

    /**
     * 扫描指定列族的数据
     * @param table
     * @throws Exception
     */
    private static void scanWithFamily(Table table) throws Exception {

        Scan scan = new Scan();
        //设置条件
        //scan.addFamily(Bytes.toBytes("f1"));
        //扫描指定列族的一个或多个属性
        scan.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"));

        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        printData(iterator);

    }

    /**
     * 扫描数据并且按照属性值进行过滤
     * @param table
     * @throws Exception
     */
    private static void scanWithColumnValue(Table table) throws Exception {

        Scan scan = new Scan();

        //列值过滤器（匹配到条件的行，整行数据都将返回）
        //SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"), CompareOperator.EQUAL, Bytes.toBytes("lili"));


        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("age"), CompareOperator.LESS, Bytes.toBytes("18"));

        scan.setFilter(filter2);

        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        printData(iterator);

    }

    /**
     * 扫描数据，并且指定多个条件
     * @param table
     * @throws Exception
     */
    private static void scanWithFilters(Table table) throws Exception {

        Scan scan = new Scan();

        //列值过滤器（匹配到条件的行，整行数据都将返回）
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"), CompareOperator.EQUAL, Bytes.toBytes("lili"));


        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("age"), CompareOperator.LESS, Bytes.toBytes("18"));

        //将多个条件组合起来
        //MUST_PASS_ALL 相当于 AND
        //MUST_PASS_ONE 相当于 OR
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, filter1, filter2);

        scan.setFilter(filterList);


        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        printData(iterator);



    }




    private static void printData(Iterator<Result> iterator) {

        while (iterator.hasNext()) {
            Result result = iterator.next();
            //获取行键
            String rowKey = Bytes.toString(result.getRow());
            System.out.println(rowKey);
            //获取列和列下面的key-value
            while (result.advance()) { //列族下有数据
                Cell cell = result.current();
                //获取cell中的key-value
                byte[] familyArray = CellUtil.cloneFamily(cell);
                String family = Bytes.toString(familyArray);

                byte[] qualifierArray = CellUtil.cloneQualifier(cell);
                String qualifier = Bytes.toString(qualifierArray);

                byte[] valueArray = CellUtil.cloneValue(cell);
                String value = Bytes.toString(valueArray);

                System.out.println(family + " , " + qualifier + " , " + value);
            }
        }
    }


}
